package com.daidai.source.mysqlsource.source;

import com.daidai.source.mysqlsource.domain.OneMarkOneSum;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class MyMysqlSource extends RichParallelSourceFunction<OneMarkOneSum> {
    private Connection connection = null;
    private PreparedStatement preparedStatement = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(
                "jdbc:mysql://localhost:3306/test",
                "root",
                "000000"
        );
        String sql = "select showName, lowMark, number, numberSum from onemarkonesum";
        preparedStatement = connection.prepareStatement(sql);
    }

    @Override
    public void run(SourceContext<OneMarkOneSum> sourceContext) throws Exception {


        ResultSet resultSet = preparedStatement.executeQuery();
        while (resultSet.next()) {
            String showName = resultSet.getNString("showName");
            int lowMark = resultSet.getInt("lowMark");
            int number = resultSet.getInt("number");
            int numberSum = resultSet.getInt("numberSum");
            sourceContext.collect(new OneMarkOneSum(showName, lowMark, number, numberSum));
        }
        Thread.sleep(1000);
    }

    @Override
    public void cancel() {
    }

    @Override
    public void close() throws Exception {
        if (connection != null) {
            connection.close();
        }
        if (preparedStatement != null) {
            preparedStatement.close();
        }
    }
}
